package com.chief.window;

import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.Map;

public class TumblingTimeWindowTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // env.setParallelism(2);

        DataStreamSource<String> streamSource = env.socketTextStream("hadoop55", 7777);

        DataStream<Tuple3<String, Date, Integer>> tuple3SingleOutputStreamOperator = streamSource.map(new MapFunction<String, Tuple3<String, Date, Integer>>() {
            @Override
            public Tuple3<String, Date, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return new Tuple3<>(split[0], DateUtils.parseDate(split[1],"yyyy-MM-dd HH:mm:ss"), Integer.valueOf(split[2]));
            }
        });

        SingleOutputStreamOperator<Tuple3<String, Date, Integer>> timestampsAndWatermarks = tuple3SingleOutputStreamOperator
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, Date, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner((h, t) -> h.f1.getTime()));

        DataStream<Tuple3<String, Date, Integer>> sum = timestampsAndWatermarks
                .keyBy(tuple3 -> tuple3.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(2);

        sum.print("TumblingTimeWindowTest sum is :");

        env.execute("TumblingTimeWindowTest");
    }
}
